﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using TUtils.BasicHelper.Logging;
using TUtils.Multithreading;
using MMessaging.MessagingEx.Common;
using TUtils.BasicHelper;
using MMessaging.MessagingEx.Queues;
using MMessaging.MessagingEx.Services;
using TUtils.Threads;

namespace MMessaging.Address.Services
{
	/// <summary>
	/// Service, der etwas ähnliches wie einen Switch realisiert.
	/// Eingehende Messages, die das Interface IMAddressMessage implementieren,
	/// werden zielgenau an den geeigneten Channel weitergeleitet.
	/// Der Service weiß, an welchen Channels welche Zieladressen liegen, weil er die 
	/// Message IMAddressAlifeMessage empfängt, die jeder adressbezogene Service periodisch
	/// sendet.
	/// Wenn die Message IMAddressAlifeMessage für eine lange Zeit ausbleibt oder der Service
	/// die Messsage IMAddressShutdownMessage empfängt, wird das Adress-Channel-Mapping 
	/// wieder gelöscht. Wenn die Message IMAddressAlifeMessage über einen Channel A 
	/// empfangen wird und ein anderer Channel B bereits für diese Adresse gemappt ist,
	/// wird das alte Mapping überschrieben.
	/// Wenn zur Adresse einer Message kein Channel gemappt ist, wird die Message
	/// in einer IMAddressNotAvailableMessage Message verpackt und an den Sender 
	/// zurückgesendet (falls dessen Adresse bekannt ist).
	/// Messages, die nicht das Interface IMAddressMessage implementieren, werden an 
	/// alle Channel weitergeleitet (Broadcast).
	/// Addressbezogene Messages, die als Receiver-Adresse MDefaultAddress.BROADCAST
	/// enthalten, werden auch an alle Channel weitergeleitet.
	/// Wenn die Zieladresse einer Message keine konkrete Serviceadresse meint, sondern eine
	/// Rollenadresse, so wird die Message an alle Services weitergeleitet, die sich
	/// in ihrer AlifeMessage für diese Rolle eingetragen haben.
	/// </summary>
	/// <typeparam name="MESSAGETYPE"></typeparam>
	/// <typeparam name="NOTAVAILABLEMESSAGE">
	/// the type of the message, which implements the interface 
	/// <code><![CDATA[
	/// IMAddressNotAvailableMessage<MESSAGETYPE>
	/// ]]></code>
	/// </typeparam>
	public class MServiceSwitch<MESSAGETYPE, NOTAVAILABLEMESSAGE>
		: TThreadX, IMService<MESSAGETYPE>, IMNodeDuplex<MESSAGETYPE>
		where MESSAGETYPE : class
		where NOTAVAILABLEMESSAGE : MESSAGETYPE, IMAddressNotAvailableMessage<MESSAGETYPE>
	{
		#region types

		private class ServiceAdressData : MAddress
		{
			/// <summary>
			/// Alle Rollen, die der Service spielt.
			/// Darf nichts enthalten. darf null zurückgeben.
			/// </summary>
			public IEnumerable<MAddress> Roles { get; set;  }

			public ServiceAdressData(
				MAddress address,
				IEnumerable<MAddress> roles
				) : base(address.AddressData)
			{
				Roles = roles;
			}
		}

		/// <summary>
		/// Delegate für eine Methode, die eine Message erzeugt, die das Interface
		/// <code><![CDATA[
		/// IMAddressNotAvailableMessage<MESSAGETYPE> 
		/// ]]></code> implementiert.
		/// </summary>
		/// <param name="undeliverableMessage"></param>
		/// <param name="receiver"></param>
		/// <param name="sender"></param>
		/// <param name="sentTime"></param>
		/// <param name="messageID"></param>
		/// <returns></returns>
		public delegate NOTAVAILABLEMESSAGE CreateNotAvailableMessageFunc(
			MESSAGETYPE undeliverableMessage,
			MAddress receiver,
			MAddress sender,
			DateTime sentTime,
			Guid messageID
			);

		#endregion

		#region private fields

		//private MChannelInterThreadPrio<MESSAGETYPE> _loopbackQueue;

		/// <summary>
		/// Die Channel-Enden, die mit dem Switch verbunden sind
		/// </summary>
		private List<IMChannelEndDuplex<MESSAGETYPE>> _channels;
		/// <summary>
		/// die Map Zieladresse -> HashSet der zugeordneten Channels
		/// <code>
		/// Channel 1---- * Serviceadresse * ---- * Rolle
		///                 |
		///                 --- 1 Timeout
		/// </code>
		/// </summary>
		private Dictionary<MAddress, IMChannelEndDuplex<MESSAGETYPE>> _mapServiceaddress2Channel = new Dictionary<MAddress, IMChannelEndDuplex<MESSAGETYPE>>();
		/// <summary>
		/// Map Zieladresse -> Zeitpunkt (UTC), wenn der mit der Zieladresse verknüpfte Service spätestens
		/// ein Lebenszeichen von sich geben muss.
		/// <code>
		/// Channel 1---- * Serviceadresse * ---- * Rolle
		///                 |
		///                 --- 1 Timeout
		/// </code>
		/// </summary>
		private Dictionary<MAddress, DateTime> _mapTimeout = new Dictionary<MAddress, DateTime>();
		/// <summary>
		/// Map Rolle -> HashSet der Zieladressen
		/// <code>
		/// Channel 1---- * Serviceadresse * ---- * Rolle
		///                 |
		///                 --- 1 Timeout
		/// </code>
		/// </summary>
		private Dictionary<MAddress, HashSet<MAddress>> _mapRole2Serviceaddress = new Dictionary<MAddress, HashSet<MAddress>>();
		/// <summary>
		/// Map Zieladresse -> HashSet der Rollen
		/// <code>
		/// Channel 1---- * Serviceadresse * ---- * Rolle
		///                 |
		///                 --- 1 Timeout
		/// </code>
		/// </summary>
		private Dictionary<MAddress, ServiceAdressData> _mapServiceaddressDatas = new Dictionary<MAddress, ServiceAdressData>();

		/// <summary>
		/// Verbindung aller Channelausgänge zu einem einzigen.
		/// </summary>
		private MChannelAdapterOut<MESSAGETYPE> _inputQueue;

		private MAddress _thisAddress;

		private CreateNotAvailableMessageFunc _createNotAvailableMessage;

		private bool _started = false;

		private object _sync = new object();

		#endregion

		#region overridables

		/// <summary>
		/// kann überschrieben werden, um ein service-spezifisches Logging zu realisieren.
		/// Wenn LogXImplementor null zurückgibt, wird das standardmäßige Logging
		/// verwendet, das im Assembl LogXImplementor.dll definiert ist und projektspezifisch
		/// ersetzt werden kann.
		/// </summary>
		protected virtual ILogXImplementor LogXImplementor
		{
			get
			{
				return null;
			}
		}

		public override void StartAsync()
		{
			_started = true;
			base.StartAsync();
		}

		#endregion

		#region constructor

		/// <summary>
		/// Konstruktor
		/// </summary>
		public MServiceSwitch(CreateNotAvailableMessageFunc createNotAvailableMessage)
		{
			_createNotAvailableMessage = createNotAvailableMessage;
			_inputQueue = new MChannelAdapterOut<MESSAGETYPE>("Channel adapter");
		}

		/// <summary>
		/// Initialisiert den Switch vollständig
		/// </summary>
		/// <param name="channels"></param>
		/// <param name="startAtOnce">
		/// <param name="thisAddress">
		/// Die Adresse dieses Switches. (Derzeit sendet 
		/// die Implementation dieses Switch aber keine Alife-Messages, so dass man dem Switch
		/// keine Message senden kann.)
		/// </param>
		/// </param>
		public void Init(
			bool startAtOnce,
			MAddress thisAddress,
			params IMChannelEndDuplex<MESSAGETYPE>[] channels)
		{
			if ( _started )
				throw new ApplicationException("jh873tfkjrfsgh");

			_thisAddress = thisAddress;
			_channels = channels.ToList();
			base.Init(
				signalHandler: OnSignal,
				threadMethod: ThreadMethod,
				threadName: GetType().Name + " " + new Guid().ToString());
			_inputQueue = new MChannelAdapterOut<MESSAGETYPE>("channel adapter of switch",channels);

			//_loopbackQueue = new MChannelInterThreadPrio<MESSAGETYPE>();
			//_inputQueue.AddChannel(_loopbackQueue);

			if (startAtOnce)
				StartAsync();
		}

		//public void PostMessage(MESSAGETYPE message)
		//{
		//    _loopbackQueue.PostMessage(message);
		//}

		#endregion

		#region private methods

		private void OnSignal(IThreadContext context, SignalEnum signal, Exception e)
		{

		}

		private MESSAGETYPE GetNextMessage(out int channelIndex)
		{
			return _inputQueue.WaitForMessage(MMessagingHelper.CANCEL_WAIT_INTERVALL, out channelIndex);
		}

		private void RemoveAddressFromMapRole2Addresses(MAddress address)
		{
			foreach(var rolePair in _mapRole2Serviceaddress.ToList())
			{
				var role = rolePair.Key;
				var addresses = rolePair.Value;
				addresses.Remove(address);
				if (addresses.Count == 0)
					_mapRole2Serviceaddress.Remove(role);
			}
		}

		private void OnAlifeMessage(
			IMAddressAlifeMessage alifeMessage,
			int channelIdx)
		{
			IEnumerable<MAddress> roles = alifeMessage.Roles;
			if ( roles == null )
				roles = new MAddress[0];

			var serviceAddress = alifeMessage.Sender;
			var channel = _channels[channelIdx];
			var now = DateTime.UtcNow;

			// renew timeout
			_mapTimeout[serviceAddress] = now.AddMilliseconds(2 * alifeMessage.IntervallTime);
			// set channel in map service address -> channel
			_mapServiceaddress2Channel[serviceAddress] = channel;

			ServiceAdressData oldAddressData;
			if ( !_mapServiceaddressDatas.TryGetValue(serviceAddress,out oldAddressData))
				oldAddressData = new ServiceAdressData(serviceAddress,new MAddress[0]);

			// check, if any role has changed ?
			bool rolesChanged = false;
			var oldRoles = oldAddressData.Roles;
			if (oldRoles.Count() != roles.Count())
				rolesChanged = true;

			if ( !rolesChanged )
			{
				foreach( var oldRole in oldRoles)
				{
					if ( !roles.Any(role=>role==oldRole))
					{
						rolesChanged = true;
						break;
					}
				}
			}

			// set map service address -> roles
			_mapServiceaddressDatas[serviceAddress] = new ServiceAdressData(serviceAddress, roles);

			// roles has changed
			if ( rolesChanged )
			{
				// reset map role -> service addresses
				RemoveAddressFromMapRole2Addresses(serviceAddress);
				foreach(var role in roles)
				{
					HashSet<MAddress> addresses;
					if ( !_mapRole2Serviceaddress.TryGetValue(role, out addresses) )
					{
						addresses = new HashSet<MAddress>();
						_mapRole2Serviceaddress[role] = addresses;
						// System.Diagnostics.Debug.WriteLine("GTZZGX role " + role + " added");
					}

					addresses.Add(serviceAddress);
				}
			}
		}

		/// <summary>
		/// Liefert alle Channels, an die eine Message mit der geg, Adresse "receiver"
		/// geleitet werden muss.
		/// </summary>
		/// <param name="receiver"></param>
		/// <returns></returns>
		private IEnumerable<IMChannelEndDuplex<MESSAGETYPE>> GetChannels(MAddress receiver)
		{
			var channels = new HashSet<IMChannelEndDuplex<MESSAGETYPE>>();
			HashSet<MAddress> serviceAddresses = null;
			IMChannelEndDuplex<MESSAGETYPE> receiverChannel;
			// if receiver address is a service address
			if (_mapServiceaddress2Channel.TryGetValue(receiver, out receiverChannel))
			{
				serviceAddresses = new HashSet<MAddress>();
				serviceAddresses.Add(receiver);
			}
			// if receiver address isn't a service address
			else
			{
				_mapRole2Serviceaddress.TryGetValue(receiver, out serviceAddresses);
			}

			// if receiver address is a service address or a role address
			if (serviceAddresses != null && serviceAddresses.Count > 0)
			{
				// check for timeouts ***
				var now = DateTime.UtcNow;
				foreach( var serviceAddress in serviceAddresses.ToList())
				{
					DateTime elapseTime;
					if (_mapTimeout.TryGetValue(serviceAddress, out elapseTime))
					{
						bool elapsed = false;
#						if !DEBUG
						// don't check elapse time, when debugging code
						elapsed = elapseTime < now;
#						endif

						// .. is not elapsed
						if (!elapsed)
						{
							if (_mapServiceaddress2Channel.TryGetValue(serviceAddress, out receiverChannel))
								channels.Add(receiverChannel);
						}
						// service hasn't answered for a long time
						else
						{
							RemoveAddressFromMapRole2Addresses(serviceAddress);
							_mapServiceaddress2Channel.Remove(serviceAddress);
							_mapServiceaddressDatas.Remove(serviceAddress);
							_mapTimeout.Remove(serviceAddress);
						}
					}
				}			
			}

			return channels;
		}

		private void OnShutdownMessage(IMAddressShutdownMessage shutdownMessage)
		{
			var serviceAddress = shutdownMessage.Sender;
			RemoveAddressFromMapRole2Addresses(serviceAddress);
			_mapServiceaddress2Channel.Remove(serviceAddress);
			_mapTimeout.Remove(serviceAddress);
			_mapServiceaddressDatas.Remove(serviceAddress);
		}

		private IEnumerable<IMChannelEndDuplex<MESSAGETYPE>> GetChannelsWithout(int channelIdx)
		{
			var channelRemoved = _channels[channelIdx];
			return _channels.Where(channel => channel != channelRemoved);
		}

		private void ThreadMethod(IThreadContext context)
		{
			while (!IsCanceling)
			{
				int channelIdx;
				var msg = GetNextMessage(out channelIdx);

				lock (_sync)
				{
					IMAddressMessage addressMsg = msg as IMAddressMessage;

					// if message contains sender and receiver address
					if (addressMsg != null)
					{
						// if message is an alife signal of a service
						if (addressMsg is IMAddressAlifeMessage)
						{
							OnAlifeMessage(addressMsg as IMAddressAlifeMessage, channelIdx);
						}
							// if message is shutdown signal of a service
						else if (addressMsg is IMAddressShutdownMessage)
						{
							OnShutdownMessage(addressMsg as IMAddressShutdownMessage);
						}

						// if message is a broadcast message
						if (addressMsg.Receiver.Equals(MDefaultAddress.BROADCAST))
						{
							// broadcasting
							foreach (var channel in GetChannelsWithout(channelIdx))
								channel.PostMessage(msg);
						}
							// if message isn't a broadcast message
						else
						{
							var receiverChannels = GetChannels(addressMsg.Receiver);

							// deliver message
							foreach (var receiverChannel in receiverChannels)
								receiverChannel.PostMessage(msg);

							// if message couldn't be delivered
							if (receiverChannels.Count() == 0)
							{
								// send message back to sender
								_channels[channelIdx].PostMessage(
									_createNotAvailableMessage(
										msg,
										addressMsg.Sender,
										_thisAddress,
										DateTime.UtcNow,
										Guid.NewGuid()));
							}
						}
					}
						// if message doesn't contain sender and receiver address
					else
					{
						if (msg != null)
						{
							// broadcasting
							foreach (var channel in GetChannelsWithout(channelIdx))
								channel.PostMessage(msg);
						}
					}
				}
			}
		}

		#endregion

		#region public

		/// <summary>
		/// May be called, even if switch has been started allready.
		/// </summary>
		/// <param name="channelEnd"></param>
		public void AddDuplexChannelEnd(IMChannelEndDuplex<MESSAGETYPE> channelEnd)
		{
			lock (_sync)
			{
				_channels.Add(channelEnd);
				_inputQueue.AddChannel(channelEnd);
			}
		}

		public string ServiceName
		{
			get
			{
				return this.ThreadName ?? string.Empty;
			}
		}

		#endregion
	}


	public class MServiceSwitch : MServiceSwitch<MAddressMessageBase, MAddressNotAvailableMessage>
	{
		public MServiceSwitch() : base(MAddressNotAvailableMessage.Create)
		{

		}
	}
}
